#include "lbserver.h"

void *reqHandler(void *arg){
	reqHandlerArg * _reqhandler = (reqHandlerArg*) arg;
	LBServer * lb = _reqhandler->_server;
	int sockfd = _reqhandler->_sockfd;
	lb->messageHandler(sockfd);
	return NULL;
}
void *spreadThread(void *arg){
	LBServer *lb = (LBServer*)arg;
	lb->resourceSpreadHandler();
	return NULL;
}

void *statisticalThread(void *arg){
	LBServer *lb = (LBServer *)arg;
	lb->doStatistical();
	return NULL;
}

void *resetThread(void *arg){
	LBServer *lb = (LBServer *)arg;
	lb->resetAllCount();
	return NULL;
}

void *rsRequestFile(void *arg){
	pthread_detach(pthread_self());
	spreadHandlerArg * spreadHandler = (spreadHandlerArg*)arg;
	LBServer *lb = spreadHandler->_server;
	lb->requestFile(spreadHandler->_sourceRs,spreadHandler->_fileId,spreadHandler->_destRs);
	//lb->deleteSpreadMessage(spreadHandler->_sourceRs,spreadHandler->_fileId,spreadHandler->_destRs);	
	pthread_exit(NULL);
	return NULL;
}

LBServer::LBServer(ServerConfigType *serverConfig){
	_serverConfig = serverConfig;
	_resourceNumber = serverConfig->resourceNumber;
	_rsNumber = serverConfig->rsNumber;
	pthread_mutex_init(&_lbmutex,NULL);
	pthread_cond_init(&_needSpread,NULL);
	resourceManager = new ResourceManager(this,serverConfig->resourceNumber,serverConfig->maxFileLen,serverConfig->minFileLen);
	clientNumber = 0;
}

LBServer::~LBServer(){
	pthread_mutex_destroy(&_lbmutex);
	pthread_cond_destroy(&_needSpread);
}

void
LBServer::run(){
	initRSList();
	resourceManager->initialLayOut(_serverConfig->rsNumber,1);
	cout<<"print the information of rs list"<<endl;
	printRSInfo();
	cout<<"create the statistical thread..."<<endl;
	pthread_t tid;
	//start statistical thread
	pthread_create(&tid,NULL,statisticalThread,this);
	pthread_detach(tid);
	//start spread thread
	cout<<"create the spread thread..."<<endl;
	pthread_create(&tid,NULL,spreadThread,this);
	pthread_detach(tid);
	//start reset thread
	pthread_create(&tid,NULL,resetThread,this);
	//start content service thread
	cout<<"start content service..."<<endl;
	startContentService(_serverConfig->serverPort);
	
}

void
LBServer::printRSInfo(){
	for(unsigned int i = 0;i<_rsNumber;i++){
		_rsList[i]->printRSInfo();
		cout<<endl;
	}
	cout<<endl;
}
void
LBServer::initRSList(){
	RS *newrs;
	unsigned int maxLoad;
	for(unsigned int i = 0;i<_rsNumber;i++){
		maxLoad = Randomi(_serverConfig->infMaxRsLoad,_serverConfig->subMaxRsLoad);
		newrs = new RS(this,i,_serverConfig->rsBandwidth,maxLoad,_serverConfig->spreadThreshold);
		_rsList.push_back(newrs);
	}
}
void
LBServer::doStatistical(){
	ofstream ofs(_serverConfig->statisticalResultFile.c_str());
	unsigned int totalLoad ;
	unsigned int currentLoad ;
	while(true){
		totalLoad = 0;
		for(unsigned int i = 0;i<_rsNumber;i++){
			currentLoad = _rsList[i]->getCurrentLoad();
			totalLoad += currentLoad;
			ofs<<currentLoad<<"\t";
		}
		ofs<<"\t"<<totalLoad<<endl;
		sleep(2);
	}
	ofs.close();

}
/*
int
LBServer::initializeFileList(unsigned int resourceNumber,unsigned int minFileLen,unsigned int maxFileLen){
	unsigned int fileLen;
	FileInfo *file;
	for(int i =0;i<resourceNumber ;i++){
		fileLen = RandomI(minFileLen,maxFileLen);
		file = new FileInfo(i+1,fileLen);
		_fileList.push_back(file);
	}
	return 0;
}

int 
LBServer::fileLayOut(unsigned int resourceNumber,unsigned int rsNumber){
	return 0;
}
*/
void
LBServer::resetAllCount(){
	while(true){
		for(unsigned int i = 0;i< _resourceNumber;i++){
			resourceManager->resetCount(i);
		}
		sleep(_serverConfig->dwTimeInterval);
	}
}

int
LBServer::addNewFile(unsigned int rsId,unsigned int fileId){
	if(_rsList.size()<= rsId){
		cout<<"wrong rsId ,size = "<<_rsList.size()<<",rsId = "<<rsId<<endl;
		exit(1);
	}
	return _rsList[rsId]->addNewFile(fileId);
}
void
LBServer::increaseClientNumber(){
	pthread_mutex_lock(&_lbmutex);
	clientNumber ++;
	pthread_mutex_unlock(&_lbmutex);
	
}

unsigned int
LBServer::getClientNumber(){
	unsigned int result;
	pthread_mutex_lock(&_lbmutex);
	result = clientNumber ;
	pthread_mutex_unlock(&_lbmutex);
	return result;
}
int
LBServer::startContentService(string serverPort){
	int listen_fd = tcp_listen(serverPort.c_str());
	int sockfd;
	pthread_t tid;
	struct sockaddr_in cliaddr;
	socklen_t cliLen = sizeof(cliaddr);
	for(;;){
		sockfd = accept(listen_fd,(struct sockaddr*)&cliaddr,&cliLen);
		if(sockfd < 0){
			perror("accept a client connection error");
			continue;
		}else{
//			increaseClientNumber();
			clientNumber ++;
			reqHandlerArg *arg = new reqHandlerArg(this,sockfd);
			pthread_create(&tid,NULL,reqHandler,(void*)arg);
			pthread_detach(tid);
		}
	}
	return 0;
}


void
LBServer::messageHandler(int sockfd){
	ClientRequestPack requestPack;
	readn(sockfd,(void*)&requestPack,sizeof(ClientRequestPack));
	if(ntohl(requestPack.messageType) != REQUESTRESOURCE && ntohl(requestPack.rsId) !=0){
		cout<<"receive REQUESTRESOURCE error !"<<endl;
		exit(1);
	}
	unsigned int clientId;
	clientId = ntohl(requestPack.clientId);
	cout<<"client "<<clientId<< " request file "<<ntohl(requestPack.fileId)<<endl;
	while(true){
		unsigned int rsId = findBestRs(ntohl(requestPack.fileId));
		cout<<"client "<<clientId<<" will request from rs "<<rsId<<endl;
		rsId = ntohl(rsId);
		writen(sockfd,(void*)&rsId,sizeof(int));
		readn(sockfd,(void*)&requestPack,sizeof(ClientRequestPack));
		//cout<<"messageType = "<<requestPack.messageType<<endl;
		if(ntohl(requestPack.messageType) == REQUESTRS ){
			cout<<"redirect the request of client "<<clientId<<" to rs "<<htonl(rsId)<<endl;
			if(downloadFile(clientId,ntohl(requestPack.rsId),ntohl(requestPack.fileId),ntohl(requestPack.clientBandwidth)) == 0){
				requestPack.messageType = htonl(REQUESTRESOURCEOVER);
				writen(sockfd,(void*)&requestPack,sizeof(requestPack));
				break;
			}
		}else{
			cout<<"receive REQUESTRESOURCE error !"<<endl;
			exit(1);
		}	
	}
	pthread_exit(NULL);
}


unsigned int
LBServer::findBestRs(unsigned int fileId){
	unsigned int resultRs;
	unsigned int minLoad = 0;
	vector<unsigned int> rsset;
	getRSSet(fileId,rsset);
	//cout<<"get the rs set of file "<<fileId<<endl;
	vector<unsigned int> ::iterator it;
	if(rsset.size() == 0){
		cout<<"program wrong"<<endl;
		exit(1);
	}
	resultRs = rsset[0];
	for(it = rsset.begin();it!=rsset.end();it++){
		cout<<"rsId = "<<*it<<endl;
		if(*it >= _rsList.size())
			exit(1);
		unsigned int moreclients =  _rsList[*it]->canAffordClients();
		cout<<"moreclients = "<<moreclients<<",minLoad = "<<minLoad<<endl;
		if(moreclients > minLoad){
			minLoad = moreclients;
			resultRs = *it;
		}
		if(minLoad == 0){
			cout<<"too much load,now have served "<<clientNumber<<endl;
			printRSInfo();
			exit(1);
		}
	}
	return resultRs;
}


int
LBServer::downloadFile(unsigned int clientId,unsigned int rsId,unsigned int fileId,unsigned int clientBandwidth){
	unsigned int fileLen = resourceManager->getFileLen(fileId);
	resourceManager->visited(fileId);
	if(rsId >= _rsList.size()){
		cout<<"wrong rsId"<<endl;
		exit(1);
	}
	return _rsList[rsId]->downloadFile(clientId,fileLen,clientBandwidth);
}




bool
LBServer::checkSpreadMessage(SpreadMessageInfo *spreadMessage){
	vector< SpreadMessageInfo *> ::iterator it;
	pthread_mutex_lock(&_lbmutex);
	for(it = spreadMessageQueue.begin();it != spreadMessageQueue.end();it++){
		if( spreadMessage->_fileId == (*it)->_fileId &&  spreadMessage->_sourceRs == (*it)->_sourceRs && spreadMessage->_destRs == (*it)->_destRs){
			pthread_mutex_unlock(&_lbmutex);
			return true;
		}
	}
	pthread_mutex_unlock(&_lbmutex);
	return false;
}

bool
LBServer::isInSpreadFile(unsigned int spreadFileId){
	vector< SpreadMessageInfo *> ::iterator it;
	pthread_mutex_lock(&_lbmutex);
	for(it = spreadMessageQueue.begin();it != spreadMessageQueue.end();it++){
		if( spreadFileId == (*it)->_fileId){
			pthread_mutex_unlock(&_lbmutex);
			return true;
		}
	}
	pthread_mutex_unlock(&_lbmutex);
	return false;
	
}

bool
LBServer::isInSpreadSource(unsigned int sourceRs){
	vector< SpreadMessageInfo *> ::iterator it;
	pthread_mutex_lock(&_lbmutex);
	for(it = spreadMessageQueue.begin();it != spreadMessageQueue.end();it++){
		if( sourceRs == (*it)->_sourceRs){
			pthread_mutex_unlock(&_lbmutex);
			return true;
		}
	}
	pthread_mutex_unlock(&_lbmutex);
	return false;

}
bool
LBServer::isInSpreadDest(unsigned int destId){
	vector< SpreadMessageInfo *> ::iterator it;
	pthread_mutex_lock(&_lbmutex);
	for(it = spreadMessageQueue.begin();it != spreadMessageQueue.end();it++){
		if( destId == (*it)->_destRs){
			pthread_mutex_unlock(&_lbmutex);
			return true;
		}
	}
	pthread_mutex_unlock(&_lbmutex);
	return false;
}
unsigned int
LBServer::findBestDest(vector<bool>& canChooseRs,unsigned int spreadFileId){
	unsigned int moreLoad,max ;
	moreLoad = max = 0;
	unsigned int resultRs = _rsNumber + 1;
	for(unsigned int i  =0;i<_rsNumber;i++){
		if(canChooseRs[i] == false)
			continue;
		if(_rsList[i]->haveFile(spreadFileId) == true){
			canChooseRs[i]=false;
			continue;
		}
		moreLoad =_rsList[i]->canAffordClients();
		if(moreLoad > max){
			max = moreLoad;
			resultRs = i;
		}
//		cout<<"rs "<<i<<" : moreLoad = "<<moreLoad<<" ,max = "<<max<<",resultRs = "<<resultRs<<endl;
	}
	return resultRs;
}
int
LBServer::needSpread(unsigned int rsId,const vector<unsigned int> &fileList){
	unsigned int spreadFileId;
//	if(isInSpreadSource(rsId)==true){
//		cout<<"rs "<<rsId<<" have spreaded"<<endl;
//		return -1;
//	}
	cout<<"will spread the resource of rs "<<rsId<<endl;
	vector<bool> canChooseFile(_resourceNumber);
	vector<bool> canChooseRs(_rsNumber);
	for(unsigned int i = 0;i<_resourceNumber;i++){
		canChooseFile[i] = false;
	}
	for(size_t i = 0;i<fileList.size();i++){
		canChooseFile[fileList[i]] = true;
	}
	for(unsigned int i = 0;i<_rsNumber;i++){
		canChooseRs[i] = true;
	}
	while(true){
		spreadFileId = resourceManager->findMaxCountFile(canChooseFile);
		cout<<"spreadFileId = "<<spreadFileId<<endl;
		break;
		/*if(isInSpreadFile(spreadFileId) == false)
			break;
		else
			canChooseFile[spreadFileId] = false;
		*/
	}
	//cout<<"correct spread file id "<<spreadFileId<<endl;
	unsigned int destId;
	canChooseRs[rsId] = false;
	while(true){
		destId = findBestDest(canChooseRs,spreadFileId);
		//cout<<"destId = "<<destId<<endl;
		if(destId >= _rsNumber){
			cout<<"too much load,now have served "<<clientNumber<<endl;
			//printRSInfo();
			return -1;
		//	exit(1);
		}
		break;
		/*
		if(isInSpreadDest(destId) == false)
			break;
		else
			canChooseRs[destId] = false;
		*/
	}
	//cout<<"correct spread dest id "<<destId<<endl;
	SpreadMessageInfo *newMessage=new SpreadMessageInfo(rsId,spreadFileId,destId);
//	cout<<"add message <"<<rsId<<","<<spreadFileId<<","<<destId<<">"<<endl;
	addSpreadMessage(newMessage);
	return 0;
}
int
LBServer::addSpreadMessage(SpreadMessageInfo *spreadMessage){
	pthread_mutex_lock(&_lbmutex);
	spreadMessageQueue.push_back(spreadMessage);
	pthread_cond_signal(&_needSpread);
//	cout<<"signal the thread to wake up"<<endl;
	pthread_mutex_unlock(&_lbmutex);
	return 0;
	/*if(checkSpreadMessage(spreadMessage) == false){
		spreadMessageQueue.push_back(spreadMessage);
		pthread_cond_signal(&_needSpread);
		cout<<"signal the thread to wake up"<<endl;
		pthread_mutex_unlock(&_lbmutex);
		return 0;
	}else{
		pthread_mutex_unlock(&_lbmutex);
//		cout<<"already have this message !"<<endl;
		return -1;
	}
*/
}
int
LBServer::deleteSpreadMessage(unsigned int sourceRs,unsigned int fileId,unsigned int destRs){
	vector<SpreadMessageInfo*>::iterator it;
	pthread_mutex_lock(&_lbmutex);
	for(it = spreadMessageQueue.begin();it!=spreadMessageQueue.end();it++){
		if((*it)->_sourceRs == sourceRs && (*it)->_destRs == destRs && (*it)->_fileId == fileId){
	//		delete *it;
//			cout<<"delete message <"<<sourceRs<<","<<fileId<<","<<destRs<<">"<<endl;
			spreadMessageQueue.erase(it);
			break;
		}
	}
	pthread_mutex_unlock(&_lbmutex);
	return 0;
}

int 
LBServer::resourceSpreadHandler(){
	pthread_t tid;
	unsigned int sourceRs,fileId,destRs;
	while(true){
//		cout<<"spread handler"<<endl;
		pthread_mutex_lock(&_lbmutex);
//		cout<<"locked in handler"<<endl;
		if(spreadMessageQueue.size() == 0){
//			cout<<"sleep"<<endl;
			pthread_cond_wait(&_needSpread,&_lbmutex);
//			cout<<"wake up"<<endl;
		}
//		cout<<"wake up"<<endl;
		//have spread message in queue
		sourceRs = spreadMessageQueue[0]->_sourceRs;
		fileId = spreadMessageQueue[0]->_fileId;
		destRs = spreadMessageQueue[0]->_destRs;
		spreadHandlerArg *arg = new spreadHandlerArg(this,sourceRs,fileId,destRs);
		spreadMessageQueue.erase(spreadMessageQueue.begin());
//		cout<<"earse in handler"<<endl;
		pthread_mutex_unlock(&_lbmutex);
//		cout<<"unlocked in handler"<<endl;
		pthread_create(&tid,NULL,rsRequestFile,arg);
	}
	return 0;
}

int
LBServer::requestFile(unsigned int sourceRs,unsigned int fileId,unsigned int destRs){
	unsigned int fileLen;
	fileLen = resourceManager->getFileLen(fileId);
	unsigned int sourceBand,destBand;
	sourceBand = _rsList[sourceRs]->getBandwidth();
	destBand = _rsList[destRs]->getBandwidth();
	unsigned int  band;
	band = sourceBand  > destBand ? destBand  : sourceBand;
	unsigned int sendTime = fileLen * 8 * 1000000 / band;
//	cout<<"length = "<<fileLen<<",band = "<<band<<endl;
	cout<<"send file "<<fileId<<" from rs "<<sourceRs<<" to rs "<<destRs<<" will take "<<sendTime/1000000.0<<" secs"<<endl;
	resourceManager->resetCount(fileId);
	mysleep(sendTime);
	resourceManager->addNewRS(fileId,destRs);
	_rsList[destRs]->addNewFile(fileId);
	return 0;
}

int
LBServer::getRSSet(unsigned int fileId,vector<unsigned int>& rsset){
	resourceManager->getRSSet(fileId,rsset);
	return 0;
}
